

package backend

import (
	"math"
	"sync"

	bolt "go.etcd.io/bbolt"
)

// IsSafeRangeBucket是一种防止无意中读取重复密钥的黑客手段；
// 在一个存储桶上覆盖的内容只能在limit=1时获取，但ISSAFERAGEBUCKET 
// 已知不会覆盖任何密钥，因此范围是安全的。

// etcd对只读事务的抽象
type ReadTx interface {
	Lock()
	Unlock()
	RLock()
	RUnlock()
	// 在指定的Bucket中进行范围查找
	UnsafeRange(bucket Bucket, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
	// 遍历指定Bucket中的全部键位对
	UnsafeForEach(bucket Bucket, visitor func(k, v []byte) error) error
}

// readTx和concurrentReadTx的基本类型，以消除这些函数之间的重复
type baseReadTx struct {
	// 在读写buf中的缓存区数据时，需要获取该锁进行同步。
	mu  sync.RWMutex
	buf txReadBuffer	// 该buffer主要用来缓存Bucket与其中键值对集合的映射关系。
	txMu    *sync.RWMutex	// 在进行查询之前，需要获取该锁进行同步。
	tx      *bolt.Tx	// 该readTx实例底层封装的bolt.TX实例，即BoltDB层面的只读事务。
	buckets map[BucketID]*bolt.Bucket
	// txWg保护tx在批处理间隔结束时不会回滚，直到使用此tx的所有读取完成。
	txWg *sync.WaitGroup
}

// readTx.UnsafeForEach（）方法，该方法会遍历指定Bucket的缓存和Bucket中的全部键值对，\
// 并通过visitor回调函数处理这些遍历到的键值对
func (baseReadTx *baseReadTx) UnsafeForEach(bucket Bucket, visitor func(k, v []byte) error) error {
	dups := make(map[string]struct{})
	getDups := func(k, v []byte) error {	// 对visitor回调函数的一层包装
		dups[string(k)] = struct{}{}
		return nil
	}
	visitNoDup := func(k, v []byte) error {	// 对visitor回调函数的一层包装
		if _, ok := dups[string(k)]; ok {
			return nil
		}
		return visitor(k, v)
	}
	// 调用bucketBuffer.ForEach（）方法，边历缓存中的键位对。
	if err := baseReadTx.buf.ForEach(bucket, getDups); err != nil {
		return err
	}
	baseReadTx.txMu.Lock()	// 操作readTx.tx字段之前，需要获取txmu锁
	// 调用unsafeForEach（）函数，边历BoltDB中的键值对
	err := unsafeForEach(baseReadTx.tx, bucket, visitNoDup)
	baseReadTx.txMu.Unlock()
	if err != nil {
		return err
	}
	return baseReadTx.buf.ForEach(bucket, visitor)
}

// 主要功能就是进行范围查询
// 只能对safeRangeBucket（即名称为“key＇’的Bucket,
// 该Bucket中的key就是前面介绍的revision,value为键值对）进行真正的范围查询，
// 对其他Bucket的查询只能返回单个键值对。
func (baseReadTx *baseReadTx) UnsafeRange(bucketType Bucket, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
	// 对非法的limit值进行重新设置
	if endKey == nil {
		// 禁止单键重复
		limit = 1
	}
	if limit <= 0 {
		limit = math.MaxInt64
	}
	// 只有查询safeRangeBucket( Bp名称为key的Bucket）时，才是真正的范围查询，否则只能返回一个键位对
	if limit > 1 && !bucketType.IsSafeRangeBucket() {
		panic("do not use unsafeRange on non-keys bucket")
	}
	// 首先从缓存中查询键位对
	keys, vals := baseReadTx.buf.Range(bucketType, key, endKey, limit)
	// 检测缓存返回的键值对数量是否达到limit的限制，如采达到limit指定的上限，则直接返回缓存的查询结果
	if int64(len(keys)) == limit {
		return keys, vals
	}

	// 查找/缓存存储桶
	bn := bucketType.ID()
	baseReadTx.txMu.RLock()	// 获取txmu锁
	bucket, ok := baseReadTx.buckets[bn]
	baseReadTx.txMu.RUnlock()
	lockHeld := false
	if !ok {
		baseReadTx.txMu.Lock()
		lockHeld = true
		bucket = baseReadTx.tx.Bucket(bucketType.Name())
		baseReadTx.buckets[bn] = bucket
	}

	// 忽略丢失的存储桶，因为可能已在此批中创建
	if bucket == nil {
		if lockHeld {
			baseReadTx.txMu.Unlock()
		}
		return keys, vals
	}
	if !lockHeld {
		baseReadTx.txMu.Lock()
	}
	c := bucket.Cursor()
	baseReadTx.txMu.Unlock()	// 释放txmu锁
	// 通过unsafeRange（）函数从BoltDB中查询
	k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
	// 将查询缓存的结采与查询BlotDB的结果合并，然后返回
	return append(k2, keys...), append(v2, vals...)
}

type readTx struct {
	baseReadTx
}

func (rt *readTx) Lock()    { rt.mu.Lock() }
func (rt *readTx) Unlock()  { rt.mu.Unlock() }
func (rt *readTx) RLock()   { rt.mu.RLock() }
func (rt *readTx) RUnlock() { rt.mu.RUnlock() }

func (rt *readTx) reset() {
	rt.buf.reset()
	rt.buckets = make(map[BucketID]*bolt.Bucket)
	rt.tx = nil
	rt.txWg = new(sync.WaitGroup)
}

type concurrentReadTx struct {
	baseReadTx
}

func (rt *concurrentReadTx) Lock()   {}
func (rt *concurrentReadTx) Unlock() {}

// 锁定为no-op。创建concurrentReadTx后不需要锁定。
func (rt *concurrentReadTx) RLock() {}

// 运行锁发出concurrentReadTx结束的信号。
func (rt *concurrentReadTx) RUnlock() { rt.txWg.Done() }
